package com.atong.code.demo;

import lombok.SneakyThrows;

import java.util.concurrent.*;

/**
 * @author tongatong
 * @name CompletableFutureDemo
 * @description 异步编排demo
 * 创建异步对象：
 *  1. static CompletableFuture<Void> runAsync(Runnable runnable)
 *  2. static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
 *  3. static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
 *  4. static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
 *  四个创建方式的不同：
 *   1. 返回值不同。1-2，有返回值；3-4，没有返回值。
 *   2. 线程池。executor是线程池，2,4使用自建的线程池，1,3没有显示的线程池，使用被提供的ForkJoinPool.commonPool()执行。
 *
 * 计算完成时，回调方法
 *  1. CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action)
 *  2. CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
 *  3. CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)
 *  4. CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn)
 *  区别：
 *   （1）1 是同步的，2 3是异步的。
 *   （2）2 该线程有默认的线程池完成执行操作，3 是使用给定的异步执行工具完成操作。
 *   （3）exceptionally 只有当语句出现异常的时候，才会执行该操作。
 *
 * 线程串行化：
 *  1. CompletableFuture<Void> thenAccept(Consumer<? super T> action)
 *  2. CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
 *  3. CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)
 *  4. <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
 *  5. <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
 *  6. <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
 *  区别：
 *   （1）accept无返回值，apply有返回值。
 *   （2）Async是异步的。
 *   （3）其他也上面类似。
 *
 * 线程并行化：
 *  1. CompletableFuture<Void> thenRun(Runnable action)
 *  2. CompletableFuture<Void> thenRunAsync(Runnable action)
 *  3. CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)
 *
 *
 * 如何创建自定义线程池：
 *  new ThreadPoolExecutor();
 * 线程池的7个参数：
 *  1. 核心线程数。
 *   （1）I/O密集型。2*n，n是处理器个数。
 *   （2）CPU密集型。n+1。
 *  2. 最大线程数。最大可启用的线程数。
 *  3. 闲置时间。空闲线程空闲到一定时间后，会被关闭。
 *  4. 时间单位。闲置时间的单位。
 *  5. 阻塞队列。又称等待队列，当核心线程满的时候，其他线程会被分配到阻塞队列中。
 *  6. 线程工厂。默认即可。
 *  7. 拒绝策略。当最大线程数满了之后，其他线程会被拒绝。
 * @date 2021/9/16
 * @time 16:57
 */
public class CompletableFutureDemo {
    @SneakyThrows
    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(
                4,//核心线程数
                10,//最大线程数
                2,//闲置时间
                TimeUnit.SECONDS,//时间单位
                new ArrayBlockingQueue<>(3),//阻塞队列
                Executors.defaultThreadFactory(),//线程工厂
                new ThreadPoolExecutor.AbortPolicy()//决绝策略
        );
//        CompletableFuture<Void> completableFuture1 = CompletableFuture.runAsync(() -> {
//            System.out.println("没有返回值，没有显示线程池");
//        });
//        CompletableFuture<Void> completableFuture2 = CompletableFuture.runAsync(() -> {
//            System.out.println("没有返回值，有显示线程池");
//        }, executorService);
//        CompletableFuture<Integer> completableFuture3 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("有返回值，没有显示线程池");
//            return 1024;
//        });
//        CompletableFuture<Integer> completableFuture4 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("有返回值，有显示线程池");
//            return 1024;
//        }, executorService);
/*        CompletableFuture<Integer> completableFuture5 = CompletableFuture.supplyAsync(() -> {
            //报错，测试形参exc
//            int i = 1/0;
            System.out.println("有返回值，有显示线程池");
            return 1024;
        }, executorService).whenComplete((res, exc) -> {
            //res是返回值，exc是异常
            System.out.println("返回结果:\t"+res);
            //当函数有异常的时候，会将异常类型输出
            System.out.println("报错信息:\t"+exc);
        });*/
/*        CompletableFuture<Integer> completableFuture6 = CompletableFuture.supplyAsync(() -> {
            System.out.println("有返回值，有显示线程池");
            return 1024;
        }, executorService).whenCompleteAsync((res, exc) -> {

             //异步执行，应该会先输出返回结果，然后在输出这里面的语句
            try {
                //睡3秒，展示异步执行
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //res是返回值，exc是异常
            System.out.println("返回结果:\t"+res);
            //当函数有异常的时候，会将异常类型输出
            System.out.println("报错信息:\t"+exc);
        });*/
/*        CompletableFuture<Integer> completableFuture7 = CompletableFuture.supplyAsync(() -> {
            System.out.println("有返回值，有显示线程池");
            return 1024;
        }).whenCompleteAsync((res, exc) -> {
            //异步执行，应该会先输出返回结果，然后在输出这里面的语句.
            //这里会将这里面的线程交由指定的线程池执行。
            try {
                //睡3秒，展示异步执行
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //res是返回值，exc是异常
            System.out.println("返回结果:\t"+res);
            //当函数有异常的时候，会将异常类型输出
            System.out.println("报错信息:\t"+exc);
        }, executorService);*/
/*        CompletableFuture<Integer> completableFuture8 = CompletableFuture.supplyAsync(() -> {
            int i = 1/0;
            System.out.println("有返回值，有显示线程池");
            return 1024;
        }).whenCompleteAsync((res, exc) -> {
            //异步执行，应该会先输出返回结果，然后在输出这里面的语句.
            //这里会将这里面的线程交由指定的线程池执行。
            try {
                //睡3秒，展示异步执行
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //res是返回值，exc是异常
            System.out.println("返回结果:\t"+res);
            //当函数有异常的时候，会将异常类型输出
            System.out.println("报错信息:\t"+exc);
        }, executorService).exceptionally((i) -> {
            //之后当函数出错的时候，才会执行这里面的语句

            //返回值结果
            System.out.println("返回值：\t"+i);
           return 1025;
        });*/
/*        CompletableFuture<Integer> completableFuture9 = CompletableFuture.supplyAsync(() -> {
//            int i = 1/0;
            System.out.println("有返回值，有显示线程池");
            return 1024;
        }).thenApplyAsync((integer -> {
            //线程串行化
            //同样，也分为同步和异步，这里是以异步，自定义异步执行工具为例
            //先执行前面的函数，拿到前面的执行结果后，在执行这里面的语句
            System.out.println("异步，自定义线程池，有返回值");
            System.out.println("前面线程执行的结果：\t"+(integer++));
            return integer;
        }), executorService).whenCompleteAsync((res, exc) -> {
            //异步执行，应该会先输出返回结果，然后在输出这里面的语句.
            //这里会将这里面的线程交由指定的线程池执行。
            try {
                //睡3秒，展示异步执行
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //res是返回值，exc是异常
            System.out.println("返回结果:\t"+res);
            //当函数有异常的时候，会将异常类型输出
            System.out.println("报错信息:\t"+exc);
        }, executorService).exceptionally((i) -> {
            //之后当函数出错的时候，才会执行这里面的语句

            //返回值结果
            System.out.println("返回值：\t"+i);
            return 1025;
        });*/
/*        CompletableFuture<Void> completableFuture9 = CompletableFuture.supplyAsync(() -> {
//            int i = 1/0;
            System.out.println("有返回值，有显示线程池");
            return 1024;
        }).thenAcceptAsync((integer -> {
            //线程串行化
            //同样，也分为同步和异步，这里是以异步，自定义异步执行工具为例
            //先执行前面的函数，拿到前面的执行结果后，在执行这里面的语句
            System.out.println("异步，自定义线程池，无返回值");
            System.out.println("前面线程执行的结果：\t"+(integer++));
        }), executorService).whenCompleteAsync((res, exc) -> {
            //异步执行，应该会先输出返回结果，然后在输出这里面的语句.
            //这里会将这里面的线程交由指定的线程池执行。
            try {
                //睡3秒，展示异步执行
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //res是返回值，exc是异常
            System.out.println("返回结果:\t"+res);
            //当函数有异常的时候，会将异常类型输出
            System.out.println("报错信息:\t"+exc);
        }, executorService).exceptionally((i) -> {
            //之后当函数出错的时候，才会执行这里面的语句

            //返回值结果
            System.out.println("返回值：\t"+i);
            return null;
        });*/
        CompletableFuture<Integer> completableFuture101 = CompletableFuture.supplyAsync(() -> {
            System.out.println("有返回值，有显示线程池");
            System.out.println("completableFuture101");
            return 1024;
        }, executorService);
        CompletableFuture<Void> completableFuture102 = completableFuture101.thenRunAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(5);
                System.out.println("completableFuture102");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, executorService);
        CompletableFuture<Void> completableFuture103 = completableFuture101.thenRunAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
                System.out.println("completableFuture103");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });


//        completableFuture1.get();
//        completableFuture2.get();
//        completableFuture3.get();
//        completableFuture4.get();
//        System.out.println(completableFuture6.get());
//        System.out.println(completableFuture7.get());
//        System.out.println(completableFuture8.get());
        //这里，最后一个串行化线程没有返回值，因此获取到的值为null
//        System.out.println(completableFuture9.get());
        //这里会先输出1，然后输出3，最后输出2
        //如果是线程串行化的话，会等待8以上，并行化的话，3等待3秒，2等待5秒。
        System.out.println(completableFuture101.get());
        System.out.println(completableFuture102.get());
        System.out.println(completableFuture103.get());

        executorService.shutdown();
    }
}
